package com.lagou.work3;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.Iterator;
import java.util.Random;
/*
    flink window机制案例
 */
public class WindowsDemo {
    public static void main(String[] args) throws Exception {
        //设置flink 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //获取自定义数据源
        DataStreamSource<String> data = env.addSource(
                new SourceFunction<String>() {
                    int num = 0;
                    @Override
                    public void run(SourceContext<String> sourceContext) throws Exception {
                        while (true){
                          sourceContext.collect(String.valueOf(num));
                          num++;
                          Thread.sleep(1000);
                        }
                    }

                    @Override
                    public void cancel() {

                    }
        });

        //获取窗口
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
        SingleOutputStreamOperator<Tuple3<String,String,String>> streamOperator = data.map(
                new MapFunction<String, Tuple3<String, String, String>>() {
                    @Override
                    public Tuple3<String, String, String> map(String s) throws Exception {
                        long l = System.currentTimeMillis();
                        String dataTime = sdf.format(l);
                        Random random = new Random();
                        int rNum = random.nextInt(5);
                        return new Tuple3<>(s,dataTime,String.valueOf(rNum));
                    }
                }
        );
        KeyedStream<Tuple3<String,String,String>,String> keyedStream = streamOperator.keyBy(value -> value.f0);
        WindowedStream<Tuple3<String,String,String>,String,TimeWindow> windowedStream = keyedStream.timeWindow(Time.seconds(5));

        //操作窗口数据
        SingleOutputStreamOperator<String> outputStreamOperator = windowedStream.apply(
                new WindowFunction<Tuple3<String, String, String>, String, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow timeWindow, Iterable<Tuple3<String, String, String>> iterable, Collector<String> collector) throws Exception {
                        Iterator<Tuple3<String, String, String>> tuple3Iterator = iterable.iterator();
                        StringBuilder builder = new StringBuilder();
                        System.out.println("-------------");
                        while (tuple3Iterator.hasNext()){
                            Tuple3<String, String, String> next = tuple3Iterator.next();
                            builder.append(next.f0 + " " + next.f1 + " " + next.f2);
                        }
                        String s1 = s + " "
                                + sdf.format(timeWindow.getStart()) + " "
                                + sdf.format(timeWindow.getEnd()) + " "
                                + builder;
                        collector.collect(s1);
                    }
                }
        );

        // 输出窗口数据
        outputStreamOperator.print();
        env.execute();
    }
}
